package com.zyh.kafka.day02;

import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;

import java.util.List;
import java.util.Map;

public class MyPartitioner implements Partitioner {
    /**
     * 获取分区号的方法
     * topic 主题名
     * key key的值
     * value value的值
     * valueBytes value序列化后的值
     * cluster 保存着kafka集群信息
     */
    @Override
    public int partition(String topic, Object key, byte[] valueBytes, Object value, byte[] keyBytes, Cluster cluster) {
        //1 获取值的哈希码
        int hashCode = value.hashCode();
        //2 获取可用分区数
        List<PartitionInfo> partitionInfos = cluster.availablePartitionsForTopic(topic);
        //3 获取分区号,返回分区号
        return Math.abs(hashCode) % partitionInfos.size();
    }

    @Override
    public void close() {

    }

    /**
     * 自定义分区类创建后执行一次初始化方法
     */
    @Override
    public void configure(Map<String, ?> map) {
    }
}
